package com.wp.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;
import scala.Int;

import java.io.IOException;
import java.util.*;

/*
最完整版本的自定义数据消费
 */
public class KafkaConsumerWithOffsetControll {
    public static void main(String[] args) throws IOException {
        //连接redis服务器(在这里是连接本地的)
        Set<HostAndPort> nodes = new HashSet();
        nodes.add(new HostAndPort("192.168.100.150", 7001));
        nodes.add(new HostAndPort("192.168.100.150", 7002));
        nodes.add(new HostAndPort("192.168.100.151", 7003));
        nodes.add(new HostAndPort("192.168.100.151", 7004));
        nodes.add(new HostAndPort("192.168.100.152", 7005));
        nodes.add(new HostAndPort("192.168.100.152", 7006));
        JedisCluster cluster = new JedisCluster(nodes);

        Properties prop = new Properties();
        //bootstrap-server
        prop.put("bootstrap.servers", "192.168.100.141:9092");
        // 指定消费者组
        prop.put("group.id", "wp");
        // 指定消费位置: earliest/latest/none
//        prop.put("auto.offset.reset", "earliest");
        // 指定消费的key的反序列化方式
        prop.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        // 指定消费的value的反序列化方式
        prop.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        prop.put("enable.auto.commit", "false");
        prop.put("session.timeout.ms", "60000");
        prop.put("max.poll.interval.ms", "60000");
        KafkaConsumer<String, String> consumer = new KafkaConsumer(prop);
        consumer.subscribe(Collections.singletonList("test-offset"));

        consumer.poll(1000);
        System.out.println(consumer.subscription());
        Set<TopicPartition> assignmentset = consumer.assignment();
        Map<Integer,Long> partitionOffsetMap = new HashMap<Integer, Long>();
        for (TopicPartition partition : assignmentset) {
            int partitionId=partition.partition();
            long offset = consumer.position(partition);
            System.out.println(partitionId + "==>" + offset);
            if(cluster.exists(String.valueOf(partitionId))){
                System.out.println("找到："+partitionId);
                partitionOffsetMap.put(partitionId,Long.valueOf(cluster.get(String.valueOf(partitionId))));
            }else{
                cluster.set(String.valueOf(partitionId),Long.toString(offset));
                partitionOffsetMap.put(partitionId,Long.valueOf(cluster.get(String.valueOf(partitionId))));
            }
        }
        for(Map.Entry<Integer,Long> entry:partitionOffsetMap.entrySet()){
            consumer.seek(new TopicPartition("test-offset",entry.getKey()), entry.getValue());
        }
        try {
            Map<TopicPartition, OffsetAndMetadata> currentOffset = new HashMap();
//            while (flag) {
            for (int i = 0; i < 2; i++) {
                ConsumerRecords<String, String> records = consumer.poll(1000);
                //如果没有数据就退出
                if (records.isEmpty()) {
                    System.out.println("没有数据！");
                    break;
                }
                System.out.println("count="+records.count());
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
                    currentOffset.clear();
                    currentOffset.put(new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset(), "metadata"));
                }
                consumer.commitAsync(currentOffset, null);
                for (TopicPartition partition : assignmentset) {
                    long offset = consumer.position(partition);
                    System.out.println(partition + "==>" + offset);
                    cluster.set(String.valueOf(partition.partition()),Long.toString(offset));
                }
            }
        } catch (Exception e) {
            System.out.println("异步提交失败！");
        } finally {
            try {
                System.out.println("同步提交！");
                consumer.commitSync();
                for (TopicPartition partition : assignmentset) {
                    long offset = consumer.position(partition);
                    System.out.println(partition + "==>" + offset);
                    cluster.set(String.valueOf(partition.partition()),Long.toString(offset));
                }
            } finally {
                consumer.close();
            }
        }
        consumer.close();
        cluster.close();
    }

}
